package com.kgc.bigdata.spark.sql

import org.apache.spark.sql.SQLContext
import org.apache.spark.{SparkConf, SparkContext}

/**
  * Spark SQL用户自定义函数
  */
object UDFFunctionApp {

  def main(args: Array[String]) {
    val sparkConf = new SparkConf().setMaster("local[2]").setAppName("UDFFunctionApp")

    val sc = new SparkContext(sparkConf)
    val sqlContext = new org.apache.spark.sql.SQLContext(sc)

    val info = sc.textFile("H:/workspace/SparkProject/src/data/hobbies.txt")

    // 进行RDD到DataFrame的转换，需要手动导入一个隐式转换，否则RDD无法转换成DF
    import sqlContext.implicits._

    // 这里其实就是一个普通的，元素为case class的RDD，直接对它使用toDF()方法，即可转换为DataFrame
    val hobbyDF = info.map(_.split("\t")).map(p => Hobbies(p(0), p(1))).toDF
    hobbyDF.show
    hobbyDF.registerTempTable("hobbies")

    /**
      * 定义和注册自定义函数三步曲：
      * 1) 定义函数：自己写匿名函数(将某个字段中逗号分隔的数量统计出来)
      * 2) 注册函数：SQLContext.udf.register()
      * 3) 使用
      */
    sqlContext.udf.register("hobby_num", (s: String) => s.split(',').size)

    sqlContext.sql("select name, hobbies, hobby_num(hobbies) as hobby_num from hobbies").show()

    sc.stop
  }

  case class Hobbies(name: String, hobbies: String)

}
